#ifndef CONNECTIONMANAGER_H
#define CONNECTIONMANAGER_H

#include <cms/ExceptionListener.h>
#include <activemq/library/ActiveMQCPP.h>
#include <decaf/lang/Thread.h>
#include <decaf/lang/Runnable.h>
#include <decaf/util/concurrent/CountDownLatch.h>
#include <decaf/lang/Integer.h>
#include <decaf/lang/Long.h>
#include <decaf/lang/System.h>
#include <decaf/lang/exceptions/RuntimeException.h>
#include <activemq/core/ActiveMQConnectionFactory.h>
#include <activemq/util/Config.h>
#include <cms/Connection.h>
#include <cms/Session.h>
#include <cms/TextMessage.h>
#include <cms/BytesMessage.h>
#include <cms/MapMessage.h>
#include <activemq/cmsutil/CmsTemplate.h>
#include <cms/MessageListener.h>
#include <stdlib.h>
#include <stdio.h>
#include <iostream>
#include <memory>
#include "log4qt/logger.h"
#include "imessageconverter.h"
#include <google/protobuf/Message.h>
#include "decaf/util/concurrent/ConcurrentStlMap.h"
#include "decaf/util/concurrent/Mutex.h"
#include <decaf/util/concurrent/Concurrent.h>

//ower
#include "eventagentsupport.h"
#include "jmstemplate.h"
#include "commonconsumer.h"
#include "messageargs.h"

class CommandConfig;

class ConnectionManager : public cms::ExceptionListener, public cms::MessageListener, public EventAgentSupport<MessageArgs>
{
private:
	class MessageHolder
	{
		decaf::util::concurrent::Mutex mutex;
		google::protobuf::Message* msg;
		bool isCompletedField;
		long long msecs;
	public:
		MessageHolder(long long msecs = 30*1000):msecs(msecs), isCompletedField(false){}
		~MessageHolder()
		{

			if (msg != NULL)
			{
				try{delete msg;}catch(...){}
			}
		}

		bool isCompleted(){return isCompletedField;}
		void putData(google::protobuf::Message* data)
		{
			isCompletedField = true;
			msg = data;
		}
		google::protobuf::Message* popData()
		{
			google::protobuf::Message* tmpmsg = msg;
			msg = NULL;
			return tmpmsg;
		}
		void wait()
		{
			synchronized(&mutex)
			{
				if(isCompletedField == false)
				{
					mutex.wait(msecs);
				}
			}
		}
		void notifyAll()
		{
			synchronized(&mutex)
			{
				mutex.notifyAll();
			}
		}
	};

	cms::ConnectionFactory* factory;

	cms::Connection* connection;
	std::string name;

	std::vector<CommonConsumer*> consumers;
	JmsTemplate* cmsTemplate;
	IMessageConverter* converter;
	std::string userName;
	std::string password;
	std::string clientId;
	Log4Qt::Logger* log;

	//message for calling style
	decaf::util::concurrent::ConcurrentStlMap<std::string, MessageHolder*>  cachedMsg;

private:

	ConnectionManager( const ConnectionManager& );
	ConnectionManager& operator= ( const ConnectionManager& );
public:
	ConnectionManager(const std::string& name,cms::ConnectionFactory* factory, 
		const std::string& userName = "", const std::string& password = "", const std::string& clientId = "") ;

	virtual ~ConnectionManager();
	void close() ;
	void open();
	/*void send(const std::string& destinationName,std::string& msg) ;
	void send(const std::string& destinationName, google::protobuf::Message* msg);*/
	void send(CommandConfig* cmdCfg, google::protobuf::Message* message);
	google::protobuf::Message* call(CommandConfig* cmdCfg,  google::protobuf::Message* message);

	void listen(CommandConfig* cmdCfg);
	void stopListen(const std::string& cmdName);

	virtual void onException( const cms::CMSException& ex AMQCPP_UNUSED);
	std::string getName() const {return name;}
	cms::ConnectionFactory* const getConnectionFactory() const{return factory;}
	static cms::Destination * createDestination(const std::string& destinationName);
private:

	std::vector<CommonConsumer*>::iterator findConsumer(const std::string& cmdName);
	void cleanup();
	virtual void onMessage( const cms::Message* message ) throw();

};
#endif